package com.cupdata.zicon.jdk7concurrent.chapter2;

import java.util.LinkedList;
import java.util.Random;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * A lock may be associated with one or more conditions. These conditions are declared in the
Condition interface. The purpose of these conditions is to allow threads to have control
of a lock and check whether a condition is true or not and, if it's false, be suspended
until another thread wakes them up. The Condition interface provides the mechanisms to
suspend a thread and to wake up a suspended thread.

All the Condition objects are associated with a lock and are created using the
newCondition() method declared in the Lock interface. Before we can do any operation
with a condition, you have to have the control of the lock associated with the condition, so
the operations with conditions must be in a block of code that begins with a call to a lock()
method of a Lock object and ends with an unlock() method of the same Lock object.
When a thread calls the await() method of a condition, it automatically frees the control of
the lock, so that another thread can get it and begin the execution of the same, or another
critical section protected by that lock.

When a thread calls the signal() or signallAll() methods of a
condition, one or all of the threads that were waiting for that condition are
woken up, but this doesn't guarantee that the condition that made them
sleep is now true, so you must put the await() calls inside a while
loop. You can't leave that loop until the condition is true. While the
condition is false, you must call await() again.

You must be careful with the use of await() and signal(). If you call the await() method
in a condition and never call the signal() method in this condition, the thread will be
sleeping forever.
A thread can be interrupted while it is sleeping, after a call to the await() method, so you
have to process the InterruptedException exception.

 * @author SunYabing
 *
 */
public class MultipleConditionsLockTest {
	public static void main(String[] args) {
		FileMock mock = new FileMock(100, 10);
		Buffer buffer = new Buffer(20);
		Producer11 producer = new Producer11(mock, buffer);
		Thread threadProducer = new Thread(producer, "Producer");

		Consumer11 consumers[] = new Consumer11[3];
		Thread threadConsumers[] = new Thread[3];
		for (int i = 0; i < 3; i++) {
			consumers[i] = new Consumer11(buffer);
			threadConsumers[i] = new Thread(consumers[i], "Consumer " + i);
		}

		threadProducer.start();
		for (int i = 0; i < 3; i++) {
			threadConsumers[i].start();
		}
	}

}

class FileMock {
	private String content[];
	private int index;

	public FileMock(int size, int length) {
		content = new String[size];
		for (int i = 0; i < size; i++) {
			StringBuilder buffer = new StringBuilder(length);
			for (int j = 0; j < length; j++) {
				int indice = (int) Math.random() * 255;
				buffer.append((char) indice);
			}
			content[i] = buffer.toString();
		}
		index = 0;
	}

	public boolean hasMoreLines() {
		return index < content.length;
	}

	public String getLine() {
		if (this.hasMoreLines()) {
			System.out.println("Mock: " + (content.length - index));
			return content[index++];
		}
		return null;
	}
}

class Buffer {
	private LinkedList<String> buffer;
	private int maxSize;
	private ReentrantLock lock;
	private Condition lines;
	private Condition space;
	private boolean pendingLines;

	public Buffer(int maxSize) {
		this.maxSize = maxSize;
		buffer = new LinkedList<>();
		
		//Multi　conditions
		lock = new ReentrantLock();
		lines = lock.newCondition();
		space = lock.newCondition();
		
		pendingLines = true;
	}

	public void insert(String line) {
		lock.lock();
		try {
			while (buffer.size() == maxSize) {
				space.await();
			}
			buffer.offer(line);
			System.out.printf("%s: Inserted Line: %d\n", Thread.currentThread()
					.getName(), buffer.size());
			lines.signalAll();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
		}
	}

	public String get() {
		String line = null;
		lock.lock();
		try {
			while ((buffer.size() == 0) && (hasPendingLines())) {
				lines.await();
			}
			if (hasPendingLines()) {
				line = buffer.poll();
				System.out.printf("%s: Line Readed: %d\n", Thread
						.currentThread().getName(), buffer.size());
				space.signalAll();
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
		}
		return line;
	}

	public void setPendingLines(boolean pendingLines) {
		this.pendingLines = pendingLines;
	}

	public boolean hasPendingLines() {
		return pendingLines || buffer.size() > 0;
	}

}

class Producer11 implements Runnable {
	private FileMock mock;
	private Buffer buffer;

	public Producer11(FileMock mock, Buffer buffer) {
		this.mock = mock;
		this.buffer = buffer;
	}

	@Override
	public void run() {
		buffer.setPendingLines(true);
		while (mock.hasMoreLines()) {
			String line = mock.getLine();
			buffer.insert(line);
		}
		buffer.setPendingLines(false);
	}
}

class Consumer11 implements Runnable {
	private Buffer buffer;

	public Consumer11(Buffer buffer) {
		this.buffer = buffer;
	}

	@Override
	public void run() {
		while (buffer.hasPendingLines()) {
			String line = buffer.get();
			processLine(line);
		}
	}

	private void processLine(String line) {
		try {
			Random random = new Random();
			Thread.sleep(random.nextInt(100));
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}